package com.dz.timer;

import com.dz.utils.ConcurrentDateUtil;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nullable;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @ClassName TimerEvent1
 * @Description TODO
 * @Author zhangdong
 * @Date 2020/11/15 11:03
 * @Version 1.0
 */
public class TimerEvent1
{

    private static Logger logger = LoggerFactory.getLogger(TimerEvent1.class);

    public static void main(String[] args) throws Exception {

        final StreamExecutionEnvironment env =StreamExecutionEnvironment.createLocalEnvironment();

        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);


        DataStreamSource  streamSource = (DataStreamSource) env.addSource(new MySourceTuple3()).name("生产数据");

     KeyedStream<Tuple3<String, Long, Long> ,String>  keyedStream = streamSource.keyBy(new KeySelector<Tuple3<String, Long, Long>, String>() {
                    @Override
                    public String getKey(Tuple3<String, Long, Long> value) throws Exception {
                        return value.f0;
                    }
                });


        keyedStream.process(new KeyedProcessFunction<String, Tuple3<String, Long, Long>, String>() {

            private boolean first =true;
            private Map<String,Long> cache = new ConcurrentHashMap<>();


            private void update(String key,Long value)
            {
                cache.put(key,value);
            }

            @Override
            public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {

                logger.info("定时器触发=={}=={}",timestamp, ConcurrentDateUtil.dateFormat(timestamp,null));
                synchronized (ctx)
                {                  logger.info("定时器注册=={}=={}",timestamp+5000, ConcurrentDateUtil.dateFormat(timestamp+5000,null));

                    ctx.timerService().registerEventTimeTimer(timestamp+5000);
                    String a = cache.toString();
                    out.collect(a);}

            }

            @Override
            public void processElement(Tuple3<String, Long, Long> value, Context ctx, Collector<String> out) throws Exception {

                synchronized (ctx)
                {
                    if(first)
                    {
                        first=false;

                        logger.info("第一次注册=={}",ConcurrentDateUtil.dateFormat(value.f2,null));

                        ctx.timerService().registerEventTimeTimer(value.f2+10000);
                        logger.info("第一次注册=={}",ConcurrentDateUtil.dateFormat(value.f2,null));
                    }
                    else
                    {
                        // logger.info("更新值");
                        //不是第一次
                        if(cache.containsKey(value.f0))
                        {
                            update(value.f0,value.f1);
                        }
                        else
                        {
                            cache.put(value.f0,value.f1);
                        }
                    }
                }

            }
        }).print("输出结果");


        env.execute("timer event1 test");

    }
}
